package gossip.v2.protocol;

import gossip.v2.cluster.GossipNode;
import gossip.v2.GossipCore;
import gossip.v2.message.GossipMessage;
import gossip.v2.message.HeartbeatMessage;
import lombok.extern.slf4j.Slf4j;
import org.ricks.ioc.utils.Logger;
import org.ricks.net.extension.plugins.HeartPlugin;
import org.ricks.net.transport.AioSession;
import java.io.IOException;
import java.util.concurrent.TimeUnit;

@Slf4j
public class GossipHeartHandler extends HeartPlugin<GossipMessage> {
    private final GossipCore core;


    public GossipHeartHandler(int heartRate, int timeout, TimeUnit unit, GossipCore core) {
        super(heartRate, timeout, unit, (session, lastTime) -> {
            GossipNode gossipNode = core.getClusterManager().resolveNode(session);
            Logger.error("{} - {} - 触发了心跳超时 - {}", null,gossipNode.getNodeId(), core.getLocalNode().getNodeId(),
                    lastTime);
            session.close(true);
        });
        this.core = core;
    }

    @Override
    public void sendHeartRequest(AioSession session) throws IOException {
        long start = System.nanoTime();
        core.getTransport().sendHeartbeat(session, HeartbeatMessage.HeartbeatType.REQUEST);

        GossipNode node = core.getClusterManager().resolveNode(session);
        if (node != null) {
            core.getTopologyManager().recordLatency(
                    node.getNodeId(),
                    System.nanoTime() - start
            );
        } else {
            Logger.error("节点是空，发送心跳失败 ", null);
        }
    }

    @Override
    public boolean isHeartMessage(AioSession session, GossipMessage msg) {
        return msg instanceof HeartbeatMessage;
    }
}
